package com.morpheus.zookeeper.listener;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.morpheus.zookeeper.config.ZookeeperProperties;

/**
 * zookeeper重连机制监听器
 * 
 * @author wangbin
 */
public class ZookeeperStateListener implements ConnectionStateListener {
	private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperStateListener.class);
	private ZookeeperProperties zookeeperProperties = null;
	private CuratorFramework curatorFramework = null;

	public ZookeeperStateListener() {
	}

	public CuratorFramework connect(ZookeeperProperties zookeeperProperties) {
		this.zookeeperProperties = zookeeperProperties;
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperProperties.getBaseSleepTimeMillis(),
				zookeeperProperties.getMaxRetries());
		this.curatorFramework = CuratorFrameworkFactory.builder()
				.connectString(zookeeperProperties.getConnectionString())
				.connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMillis())
				.sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMillis()).retryPolicy(retryPolicy)
				.namespace(zookeeperProperties.getNamespace()).build();
		this.curatorFramework.getConnectionStateListenable().addListener(this);
		this.curatorFramework.start();
		return this.curatorFramework;
	}

	public CuratorFramework reinit(CuratorFramework client, ZookeeperProperties zookeeperProperties) {
		try {
			unregister(client);
			client = connect(zookeeperProperties);
		} catch (Throwable th) {
			LOGGER.error("ZookeeperStateListener.reinit() throwable", th);
		}
		return client;
	}

	public static void unregister(CuratorFramework client) {
		try {
			if (client != null) {
				client.close();
				client = null;
			}
		} catch (Throwable th) {
			LOGGER.error("ZookeeperStateListener.unregister() throwable", th);
		}
	}

	@Override
	public void stateChanged(CuratorFramework client, ConnectionState newState) {
		if (newState == ConnectionState.CONNECTED) {
			LOGGER.debug("ZookeeperStateListener.stateChanged()  connected established");
		} else if (newState == ConnectionState.LOST) {
			LOGGER.warn("ZookeeperStateListener.stateChanged()  connection lost, waiting for reconection");
			try {
				LOGGER.debug("ZookeeperStateListener.stateChanged()  reinit--->");
				this.curatorFramework = reinit(client, this.zookeeperProperties);
				LOGGER.debug("ZookeeperStateListener.stateChanged()  init done");
			} catch (Throwable th) {
				LOGGER.error("ZookeeperStateListener.stateChanged()  reinit exception", th);
			}
		}
	}
}
